package com.tgy.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/***
 * @ClassName: KafkaConsumer
 * @Description:
 * @Auther: 送你一块二向箔
 * @Date: 10:11 2022/4/7
 * @version : V1.0
 */
@Component
public class KafkaConsumer {


    //设置消费组，消费指定topic
//    @KafkaListener(topics = "yaml1", groupId = "MyGroup1")
//    public void listenGroup(ConsumerRecord<String, String> record, Acknowledgment ack) {
//        String value = record.value();
//        System.out.println(value);
//        System.out.println(record);
//        //手动提交offset
//        ack.acknowledge();
//    }

    //设置消费组、多topic、指定分区、指定偏移量消费及设置消费者个数。
//    这个注解表示监听
    @KafkaListener(groupId = "ccc", topicPartitions = {
            @TopicPartition(topic = "topic1", partitions = {"1"}),
            //从100 开始消费
            @TopicPartition(topic = "yaml", partitions = "1", partitionOffsets = @PartitionOffset(partition = "0", initialOffset = "120"))}
            , concurrency = "1")//concurrency就是同组下的消费者个数，就是并发消费数，建议小于等于分区总数
    public void listenGroup1(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String value = record.value();
        System.out.println(value);
        System.out.println(record);
        //手动提交offset
        ack.acknowledge();
    }


    @KafkaListener(groupId = "tgyTest", topicPartitions = {
            //@TopicPartition(topic = "aiops_forecast_alarm", partitions = {"1"}),
            //从100 开始消费
            @TopicPartition(topic = "aiops_forecast_alarm", partitions = "1", partitionOffsets = @PartitionOffset(partition = "0", initialOffset = "1"))}
            , concurrency = "1")//concurrency就是同组下的消费者个数，就是并发消费数，建议小于等于分区总数
    public void listenGroup2(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String value = record.value();
        System.out.println(value);
        System.out.println(record);
        //手动提交offset
        ack.acknowledge();
    }

        @KafkaListener(topics = "aiops_forecast_alarm", groupId = "MyGroup1")
    public void listenGroup(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String value = record.value();
        System.out.println(value);
        System.out.println(record);
        //手动提交offset
        ack.acknowledge();
    }


}
